#include "EventLoop.h"

EventLoop::EventLoop(Acceptor& acceptor)
:_epfd(createEpollFd())
,_evtList(1024)
,_isLooping(false)
,_acceptor(acceptor)
,_conns()
,_evtfd(createEventFd())
,_pendings()    //不能用_pendings(20)的原因是Fucntor没有默认的初始值,这会导致，初始化时size()=20;
,_mutex()
{
    //能用reserve的原因是,只是分配了内存，而没有初始化。即size()=0,capacity()=20;
    _pendings.reserve(20);  //为"任务"初始化20个空间
    //开启_acceptor的监听
    _acceptor.ready();
    //将listenfd加入epoll中监听(socket)
    addEpollReadFd(_acceptor.fd());
    //将_eventfd加入监听
    addEpollReadFd(_evtfd);
}

EventLoop::~EventLoop(){
    close(_epfd);   //关闭epoll的文件描述符 
    close(_evtfd);  //关闭通知Reactor的文件描述符
}

void EventLoop::loop(){
    _isLooping = true;
    while(_isLooping){
        waitEpollFd();
    }
}

void EventLoop::unloop(){
    _isLooping = false;
}

void EventLoop::waitEpollFd(){
    int nready = 0;
        
    do{
        nready = epoll_wait(_epfd,&*_evtList.begin()/*获取vector第一个元素的首地址*/,5000,3000);
    }while( -1 == nready && errno == EINTR);

    //如果空间快满了，就扩容
    if((size_t)nready == _evtList.size()){
        _evtList.reserve(_evtList.size()*2);
    }

    if(-1 == nready){
        perror("epoll_wait ");
        unloop();
        return ;
    }else if(0 == nready){
        /* cout << "epoll_wait timeout" << endl; */
        return;
    }
    for(int idx =0; idx < nready; ++idx){
        if(!(_evtList[idx].events & EPOLLIN)){
            continue;
        }
        if(_evtList[idx].data.fd == _acceptor.fd()){
            //如果是连接事件
            handleNewConnection();
        }else if(_evtList[idx].data.fd == _evtfd){
            //如果是通知事件(工作线程执行了wakeup)
            handleRead();
            doPendingFunctors();    //执行所有的待执行的"任务"(将结果发个客户端)
        }else{
            //不是连接事件也不是通知事件，而是读写事件(消息传递事件)
            int confd = _evtList[idx].data.fd;
            handleMessage(confd);
        }
    }
}

//处理新的连接
void EventLoop::handleNewConnection(){
    //三次握手建立成功
    int fd = _acceptor.accept();
    if(fd < 0){
        perror("handleNewConnection accept");
        exit(1);
    }
    //将新的连接加入红黑树监听
    addEpollReadFd(fd);
    
    shared_ptr<TcpConnection> conn(new TcpConnection(fd,this));
    //设置TcpConnection中的三个事件
    conn->setNewConnectionCallback(_onNewConnectionCb); //连接建立的注册
    conn->setMessageCallback(_onMessageCb); //消息到达的注册
    conn->setCloseCallback(_onCloseCb); //连接断开的注册

    //以键值对的形式存起来
    _conns[fd] = conn;

    conn->handleNewConnectionCallback(); //连接建立的事件触发时机已经到达
}

//处理老的连接上的消息
void EventLoop::handleMessage(int fd){
    auto it = _conns.find(fd);
    if(it != _conns.end()){
        //连接是存在的，可以进行数据的收发
        //read/recv = ret = 0
        bool flag = it->second->isClosed(); //判断此时连接是否断开
        if(flag){
            //连接已经断开
            it->second->handleCloseCallback(); //连接断开的事件的触发时机已经到达
            delEpollReadFd(fd); //将文件描述符从红黑树上摘除掉
            _conns.erase(it);   //同时将该链接从map中删除
            close(fd);      //关闭文件描述符
        }else{
            //连接还存在
            it->second->handleMessageCallback();    //消息到达事件的触发器时机已到
        }
    }else{
        //连接不存在，可以直接让程序退出来
        return ;
    }
}

int EventLoop::createEpollFd(){
    int epfd = epoll_create(1);
    if(-1 == epfd){
        perror("epoll_create");
        exit(1);
    }
    return epfd;
}

void EventLoop::addEpollReadFd(int fd){
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN;
    
    int ret = epoll_ctl(_epfd,EPOLL_CTL_ADD,fd,&evt);
    if(-1 == ret){
        perror("epoll_ctl add");
        exit(1);
    }
}

void EventLoop::delEpollReadFd(int fd){
    int ret = epoll_ctl(_epfd,EPOLL_CTL_DEL,fd,NULL);
    if(-1 == ret){
        perror("epoll_ctl del");
        exit(1);
    }
}

void EventLoop::setNewConnectionCallback(TcpConnectionCallback &&cb){
    _onNewConnectionCb = std::move(cb);
}  

void EventLoop::setMessageCallback(TcpConnectionCallback &&cb){
    _onMessageCb = std::move(cb);
}

void EventLoop::setCloseCallback(TcpConnectionCallback &&cb){
    _onCloseCb = std::move(cb);
}

//封装write(对于eventfd)
void EventLoop::wakeup(){
   uint64_t number = 10;
   ssize_t sret;
   sret = write(_evtfd,&number,sizeof(uint64_t));
   if(sret != sizeof(uint64_t)){
       perror("write error in wakeup");
       exit(1);
   }
} 

//将"任务"存放在容器中并执行wakeup
void EventLoop::runInLoop(Functor && cb){   //MyTask::process --> _conn->sendInLoop --> _loop->runInLoop

    //也就是说明，这个函数是在MyTask的process中调用的，而process是工作线程执行的
    unique_lock<mutex> autoMutex(_mutex);
    _pendings.push_back(std::move(cb));
    autoMutex.unlock();
    
    //数据处理好了，通知Ractor/EventLoop
    wakeup();
}

//封装read函数(对于eventfd)
void EventLoop::handleRead(){
    uint64_t number;
    ssize_t sret;
    sret = read(_evtfd,&number,sizeof(uint64_t));
    if(sret != sizeof(uint64_t)){
            perror("write error in wakeup");
    }
} 

int EventLoop::createEventFd(){
    int evtfd = eventfd(0,0);
    return evtfd;
}

//执行所有"任务"
//"任务"就是线程池处理好之后的数据以及发送数据能力TcpConnection中
//的send函数
void EventLoop::doPendingFunctors(){
    vector<Functor> temp;

    unique_lock<mutex> autoMutex(_mutex);
    swap(temp,_pendings);  //将"任务"容器中的任务都交换出来执行
    autoMutex.unlock();

    //执行所有任务
    for(auto &cb : temp){
        cb(); //执行"任务"
    }
}  
